package com.demo.hadoop;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 * @author lee9213@163.com
 * @version 1.0
 * @date 2017/3/22 13:52
 */
public class HadoopUtil {

    private static final String LOCAL_FILE_PATH_SRC = "/home/longlong/temp/aaa.f";
    private static final String LOCAL_FILE_PATH_DST = "/home/longlong/temp/bbb.f";
    private static final String HDFS_FILE_PATH = "hdfs://localhost:9000/longlong/temp/aaa.f";
    private static final String HDFS_DIRECTORY_PATH = "hdfs://localhost:9000/longlong/";

    /**
     * 将本地文件写入HDFS
     */
    private static void writeFile() throws IOException {
        Configuration configuration = new Configuration();
        Path dstFile = new Path(HDFS_FILE_PATH);
        FileSystem fs = dstFile.getFileSystem(configuration);
        fs.copyFromLocalFile(false, true, new Path(LOCAL_FILE_PATH_SRC), dstFile);
    }

    /**
     * 将HDFS上的文件读到本地
     */
    private static void readFile() throws IOException {
        Configuration configuration = new Configuration();
        Path srcFile = new Path(HDFS_FILE_PATH);
        FileSystem fs = srcFile.getFileSystem(configuration);
        fs.copyToLocalFile(srcFile, new Path(LOCAL_FILE_PATH_DST));
    }

    /**
     * 遍历HDFS目录中的文件
     */
    private static void listDirectory() throws IOException {
        Configuration configuration = new Configuration();
        Path file = new Path(HDFS_DIRECTORY_PATH);
        FileSystem fs = file.getFileSystem(configuration);
        FileStatus[] fileList = fs.listStatus(file);
        for (int i = 0; i < fileList.length; i++) {
            System.out.println("name:" + fileList[i].getPath().getName() + "/t/t size:" + fileList[i].getLen());
        }
    }

    /**
     * 删除HDFS上的文件
     *
     * @throws IOException
     */
    private static void deleteFile() throws IOException {
        Configuration configuration = new Configuration();
        Path file = new Path(HDFS_FILE_PATH);
        FileSystem fs = file.getFileSystem(configuration);
        fs.delete(file, true);//第二个参数表示递归删除
    }

    /**
     * 追加写入到HDFS上的文件
     */
    private static void appendFile() throws IOException {
        Configuration configuration = new Configuration();
        Path file = new Path(HDFS_FILE_PATH);
        FileSystem fs = file.getFileSystem(configuration);
        FSDataOutputStream fsdOut = fs.append(file);
        String outstr = "hello world";
        fsdOut.writeBytes(outstr);
        fsdOut.flush();
        fsdOut.close();
    }

}
